package org.stratusdb.funsql.execute;

import org.stratusdb.Config;
import org.stratusdb.error.Error;
import org.stratusdb.types.AbstractSimpleType;
import org.stratusdb.types.TableType;

public class EquiSelection extends UnaryOperator {

	private static final long serialVersionUID = 6313732867112964227L;

	private int selectAtt;
	private AbstractSimpleType selectLit;
	private TableType inputTable;
	private int inputCount;

	public EquiSelection(AbstractOperator child, int selectAtt,
			AbstractSimpleType selectLit) {
		super(child);

		this.selectAtt = selectAtt;
		this.selectLit = selectLit;
	}

	@Override
	public Error open() {
		this.lastError = this.child.open();

		if (this.lastError != Error.NO_ERROR) {
			return this.lastError;
		}

		this.inputTable = this.child.next();
		this.inputCount = 0;
		this.lastError = this.child.getLastError();
		return this.lastError;
	}

	@Override
	public TableType next() {
		TableType resultTable = new TableType();
		this.lastError = Error.NO_ERROR;

		while (resultTable.size() < Config.EXECUTE_MAX_FETCHSIZE
				&& this.inputTable != null) {

			if (this.inputCount < this.inputTable.size()) {
				AbstractSimpleType selectValue = inputTable.getTuple(
						this.inputCount).getValue(this.selectAtt);
				if (selectValue.equals(this.selectLit)) {
					resultTable.addTuple(inputTable.getTuple(this.inputCount));
				}
				this.inputCount++;
			} else {
				this.inputTable = this.child.next();
				this.inputCount = 0;

				if (this.lastError != Error.NO_ERROR) {
					return null;
				}
			}
		}

		// return result table
		if (resultTable.size() == 0)
			return null;
		else
			return resultTable;
	}

	@Override
	public Error close() {
		this.inputTable = null;
		this.inputCount = 0;

		return this.child.close();
	}

}
